package execute

import (
	"errors"
	"fmt"
	"github.com/RoaringBitmap/roaring/roaring64"
	"github.com/panjf2000/ants"
	"net/http"
	"net/url"
	"os"
	"sieve_engine/lib/log"
	"sieve_engine/task"
	"sieve_engine/utils"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

var (
	timeFormat            = "2006-01-02 15:04:05"
	timeFormatMillisecond = "2006-01-02 15:04:05.999999999"
	sieveExecutor         = map[string]TaskExecutorInterface{}
	poolOnce              = sync.Once{}
	antPools              *ants.PoolWithFunc
)

type TaskExecutorParams struct {
	BatchIndex  int        // 批次号
	GorCount    int        // 协程数量
	TaskInfo    *task.Info // 任务信息
	GorCounter  *atomic.Int32
	RunningGors *atomic.Int32
}

type TaskExecutorInterface interface {

	// Execute /* 输出的结果跟输入的结果数量,是一致的,顺序可能不一致 */
	Execute(params TaskExecutorParams, taskProgress *task.Result) error
}

func Get(operator string) (TaskExecutorInterface, bool) {

	executor, ok := sieveExecutor[operator]

	return executor, ok
}

func Register(operator string, taskImpl TaskExecutorInterface) {

	if _, ok := sieveExecutor[operator]; ok {
		return
	}
	sieveExecutor[operator] = taskImpl
}

type DefaultResultCache struct {
	locker sync.Mutex
	notify chan struct {
		Operator    string
		AreaCode    string
		PhoneNumber string
	}
	bitmapCache map[string]map[string]*roaring64.Bitmap // key:业务类型+国家区号，例如:fb_1表示脸书的美国筛选
}

func NewDefaultResultCache() *DefaultResultCache {

	drc := &DefaultResultCache{
		notify: make(chan struct {
			Operator    string
			AreaCode    string
			PhoneNumber string
		}, 10240),
		bitmapCache: make(map[string]map[string]*roaring64.Bitmap, 0),
	}

	go func() {

		for {
			select {
			case <-time.After(time.Hour * 1):
				drc.locker.Lock()

				for _, v := range drc.bitmapCache {
					for _, vv := range v {
						vv.Clear()
					}
				}

				drc.locker.Unlock()
			case data := <-drc.notify:
				if err := drc.add(data.Operator, data.AreaCode, data.PhoneNumber); err != nil {
					if proxyDebug {
						log.Err(fmt.Sprintf("添加缓存失败:%+v", err))
					}
				}
			}
		}
	}()

	return drc
}

func (def *DefaultResultCache) Add(operator, areaCode string, phoneNumber string) error {

	select {
	case def.notify <- struct {
		Operator    string
		AreaCode    string
		PhoneNumber string
	}{Operator: operator, AreaCode: areaCode, PhoneNumber: phoneNumber}:
	default:
		break
	}

	return nil
}

func (def *DefaultResultCache) add(operator, areaCode string, phoneNumber string) error {

	if b := def.locker.TryLock(); !b {
		return errors.New("try lock failed")
	}
	defer def.locker.Unlock()

retry:
	if v, ok := def.bitmapCache[operator]; !ok {
		def.bitmapCache[operator] = make(map[string]*roaring64.Bitmap)
		goto retry
	} else {
		if _, ok := v[areaCode]; !ok {
			def.bitmapCache[operator][areaCode] = roaring64.NewBitmap()
			goto retry
		}
	}

	x, err := strconv.ParseUint(phoneNumber[len(areaCode):], 10, 64)
	if err != nil {
		return err
	}

	def.bitmapCache[operator][areaCode].Add(x)

	return nil
}

func (def *DefaultResultCache) Exist(operator, areaCode string, phoneNumber string) bool {

	if b := def.locker.TryLock(); !b {
		return false
	}
	defer def.locker.Unlock()
	if v, ok := def.bitmapCache[operator]; !ok {
		return false
	} else {
		if _, ok := v[areaCode]; !ok {
			return false
		}
	}

	x, err := strconv.ParseUint(phoneNumber[len(areaCode):], 10, 64)
	if err != nil {
		return false
	}

	return def.bitmapCache[operator][areaCode].Contains(x)
}

type DefaultExecutor struct {
	Debug                bool
	NetworkErrRetryCount int
	PrintMore            bool
	ProxyEnable          bool
	ProxyDebug           bool
	*DefaultResultCache
}

/*
没有 + ,则加上加号，否则不变
*/
func (s *DefaultExecutor) getPhone(phone string) string {

	p := strings.TrimPrefix(phone, "+")
	number := "+" + p
	return number
}

type gorParams struct {
	phones       []string
	taskProgress *task.Result
	params       *TaskExecutorParams
}
type fExecutorCallBack func(phone string, params TaskExecutorParams, roundTripper http.RoundTripper, reuseProxy bool) (http.RoundTripper, bool, *url.URL, error)

func (s *DefaultExecutor) Execute(params TaskExecutorParams, taskProgress *task.Result, fCallBack fExecutorCallBack) error {

	var (
		phones      = taskProgress.Phones()
		retryCount  = s.NetworkErrRetryCount // 失败这个多次就设置为失败(筛以有效设置为invalid?) -1 表示一直重试
		finishCount = atomic.Int32{}
		gorCount    = params.GorCount
		taskInfo    = params.TaskInfo
		poolOption  = []ants.Option{
			ants.WithExpiryDuration(time.Duration(1800) * time.Second), // 30分钟
			ants.WithPreAlloc(true),
			ants.WithPanicHandler(func(err interface{}) {
				log.Err(fmt.Sprintf("执行任务崩溃:%+v", utils.Stack()))
				_, _ = fmt.Fprintln(os.Stderr, err)
			}),
		}
		poolCallBackDebug = func(i interface{}) {
			var (
				pp                = i.(*gorParams).phones[0]
				localTaskProgress = i.(*gorParams).taskProgress
			)
			localTaskProgress.AppendResult(fmt.Sprintf("%v", pp), true)
		}
		poolCallBack = func(i interface{}) {

			var (
				isValid           bool
				index             int
				start             = time.Now()
				reuseProxy        = false
				roundTripper      http.RoundTripper
				proxyUrl          *url.URL
				pp                = i.(*gorParams).phones[0]
				localTaskProgress = i.(*gorParams).taskProgress
				localParams       = i.(*gorParams).params
			)

			p := s.getPhone(pp)
			defer func() {
				params.GorCounter.Add(-1)
				params.RunningGors.Add(-1)

				if s.PrintMore {
					end := time.Now()
					log.Info(fmt.Sprintf("当前(任务编号:%v,批次号:%v,协程编号:%v,号码:%v),任务情况:(总任务个数:%v,分配 %v 个协程任务,执行的是 %v 个料子,已经完成任务数:%v), 使用代理:(%v) 退出循环，执行任务耗时(耗时: %v,开始时间:%v 结束时间:%v)",
						taskInfo.Id, params.BatchIndex, index, p, len(phones), gorCount, index, finishCount.Add(1), proxyUrl,
						end.Sub(start).String(), start.Format(timeFormatMillisecond), end.Format(timeFormatMillisecond)))
				}
			}()

			for j := 0; j < retryCount || retryCount == -1; j++ {

				if localParams.TaskInfo.State == task.StatusPause || localParams.TaskInfo.State == task.StatusTerminate {

					if s.PrintMore {
						log.Info(fmt.Sprintf("任务:%v 被暂停或者被终止", localParams.TaskInfo.Id))
					}

					localTaskProgress.FastFinish()
					return
				}

				var pInfo proxyInfo
				if getProxyCount() > 200 {
					if pInfo = getProxy(); pInfo.str != "" {
						roundTripper = pInfo.rt
						reuseProxy = true
						if s.PrintMore {
							log.Info(fmt.Sprintf("有效代理的缓冲数量:%v", getProxyCount()))
						}
					}
				}

				reqStart := time.Now()
				if rt, b, pUrl, err := fCallBack(p, params, roundTripper, reuseProxy); err != nil {

					if err == RequestErrorForbidden {
						if bb := s.Exist(taskInfo.Operator, taskInfo.AreaCode, pp); bb {
							setProxy(fmt.Sprintf("%v", pUrl), rt)
							isValid = b
							proxyUrl = pUrl
						} else {
							isValid = false
						}
						break
					}

					_reuseProxy := reuseProxy
					if reuseProxy {
						reuseProxy = false
						delProxy(fmt.Sprintf("%v", pInfo.str))
					}

					sleepDuration := time.Millisecond * 3
					if s.PrintMore {
						reqEnd := time.Now()
						log.Err(fmt.Sprintf("当前(任务编号:%v,批次号:%v,协程编号:%v,号码:%v,耗时:%v 执行的是第%v个料子),使用代理:(%v %v) 执行任务(第%v次 )失败:%v 休息%v后继续执行",
							taskInfo.Id, params.BatchIndex, index, p, reqEnd.Sub(reqStart).String(), index, pUrl, _reuseProxy, j+1, err, sleepDuration))
					}
					time.Sleep(sleepDuration)
					if checkNetError(err) || checkDeadlineExceededError(err) || checkIPError(err) {
						reuseProxy = false
						continue
					}

				} else {
					if err = s.Add(taskInfo.Operator, taskInfo.AreaCode, pp); err != nil {
						if s.PrintMore {
							log.Err("添加缓存失败:%+v", err)
						}
					}
					/* 没有报错，说明代理使用情况挺好的，继续使用 */
					setProxy(fmt.Sprintf("%v", pUrl), rt)
					isValid = b
					proxyUrl = pUrl
					break
				}
			}

			var v = pp
			if isValid {
				v = fmt.Sprintf("%v", pp)
			}

			localTaskProgress.AppendResult(v, isValid)

			return

		}
	)

	poolOnce.Do(func() {
		antPools, _ = ants.NewPoolWithFunc(50000, func(i interface{}) {
			for _, ii := range i.(*gorParams).phones {
				data := &gorParams{
					phones:       []string{ii},
					taskProgress: i.(*gorParams).taskProgress,
					params:       i.(*gorParams).params,
				}

				if s.Debug {
					poolCallBackDebug(data)
				} else {
					poolCallBack(data)
				}
			}
		}, poolOption...)
	})

	var (
		minGors      = 500
		maxChunkSize = 200
	)

	gors := params.TaskInfo.TaskCount / minGors
	if gors == 0 {
		gors = params.TaskInfo.TaskCount % minGors
	} else {
		if gors < minGors {
			gors = minGors
		}
	}

	chunkSize := params.TaskInfo.TaskCount / gors
	if chunkSize == 0 {
		chunkSize = params.TaskInfo.TaskCount % gors
	}
	if chunkSize > maxChunkSize {
		chunkSize = maxChunkSize
	}

	if s.Debug {

		params.GorCounter.Add(-int32(params.GorCount))
		params.RunningGors.Add(-int32(params.GorCount))
		data := &gorParams{
			phones:       taskProgress.Phones(),
			taskProgress: taskProgress,
		}

		if err := antPools.Invoke(data); err != nil {
			log.Err(fmt.Sprintf("提交任务失败"))
		}
		time.Sleep(time.Second * 5)
		return nil
	}

	// chunkSize 是每组多少个元素
	ss := utils.ChunkSliceString(phones, chunkSize)
	for i := 0; i < len(ss); i++ {

		data := &gorParams{
			phones:       ss[i],
			taskProgress: taskProgress,
			params:       &params,
		}

		if err := antPools.Invoke(data); err != nil {
			log.Err(fmt.Sprintf("提交任务失败"))
			continue
		}
	}

	return nil
}
